package com.xxl.job.executor.util;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.executor.dto.ApiInfoDto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.math.BigDecimal;
import java.net.URI;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author dingfeng
 * @version 1.0
 * @DEC SQL拼装
 * @date 2022/3/3 15:03
 */
@Component
@Slf4j
public class TaskUtil {
    @Value("${api.info.mode}")
    private String api_info_mode;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private PlatformTransactionManager transactionManager;

    //处理线程池
    private ExecutorService executorService = Executors.newFixedThreadPool(2);

    /**
     * 处理方法
     *
     * @param params 调度中心传过来的参数 包含模式名 + 表名
     * @return
     */
    public void handler(String params) throws Exception {
        try {
            Future<Boolean> submit = executorService.submit(new Task(params));
            submit.get();
        } catch (Exception e) {
            //抛出异常 让调度中心获取异常信息
            throw e;
        }
    }

    /**
     * 分发批量插入请求
     * @param apiInfoDto
     * @throws Exception
     */
    public  void distributeHttpGet(ApiInfoDto apiInfoDto) throws Exception{
        /**
         * 先发送一次请求
         * 1.可以判断接口是否通
         * 2.获取到表的数据总量 分批插入
         */
        HashMap<String, String> requestParams = new HashMap<>();
        requestParams.put("pageNumber","1");
        requestParams.put("pageSize","1");
        String res = httpGet(apiInfoDto.getApiPath(), requestParams);
        JSONObject parseObject = JSONObject.parseObject(res);
        Map<String, Object> innerMap = parseObject.getInnerMap();
        JSONObject o = (JSONObject)innerMap.get("data");
        //数据总数
        Integer totalNum = (Integer)o.get("totalNum");
        //阿里中台有限制，一次最多只返回2000条数据
        Integer distributeNumber = totalNum % 2000 == 0 ? totalNum / 2000 : totalNum / 2000 + 1;
        System.out.println("distributeNumber = " + distributeNumber);
        //初始化子任务线程池
        ExecutorService distributeExecutorService = Executors.newFixedThreadPool(distributeNumber);
        //主线程 栅栏
        CountDownLatch mainThreadLatch = new CountDownLatch(1);
        //子线程 栅栏
        CountDownLatch workerThreadLatch = new CountDownLatch(distributeNumber);
        //回滚标志
        AtomicBoolean rollbackFlag = new AtomicBoolean(false);
        //返回结果
        ArrayList<Future> allRes = new ArrayList<>(distributeNumber);
        for (Integer i = 0; i < distributeNumber; i++) {
            ApiInfoDto tempApiInfoDto = new ApiInfoDto();
            BeanUtils.copyProperties(apiInfoDto,tempApiInfoDto);
            tempApiInfoDto.setPageNum(i+1);
            Future<TaskReturn> result = distributeExecutorService.submit(new DistributeTask(mainThreadLatch,workerThreadLatch,rollbackFlag,tempApiInfoDto));
            allRes.add(result);
        }
        Thread.sleep(1000);
        int threadCount = ((ThreadPoolExecutor)distributeExecutorService).getActiveCount();
        System.out.println("threadCount = " + threadCount);
        workerThreadLatch.await();
        mainThreadLatch.countDown();
        for (Future<TaskReturn> allRe : allRes) {
            TaskReturn taskReturn = allRe.get();
            System.out.println(taskReturn.toString());
        }
        //结束线程池
        distributeExecutorService.shutdown();

    }
    /**
     * 处理htppGet请求
     */
    public static String httpGet(String apiPath,Map<String,String> params) throws Exception {
        CloseableHttpClient httpclient = HttpClients.createDefault();
        URIBuilder uriBuilder = new URIBuilder(apiPath);
        // 定义请求的参数
        if(null != params){
            Iterator<String> iterator = params.keySet().iterator();
            while (iterator.hasNext()){
                String key = iterator.next();
                uriBuilder.setParameter(key,params.get(key));
            }
        }
        URI uri = uriBuilder.build();
        HttpGet request = new HttpGet(uri);
        request.setHeader("accept", "*/*");
        CloseableHttpResponse response = null;
        try {
            // 执行请求
            response = httpclient.execute(request);
            // 判断返回状态是否为200
            if (response.getStatusLine().getStatusCode() == 200) {
                //请求体内容
                String content = EntityUtils.toString(response.getEntity(), "UTF-8");
                //内容写入文件
                return content;
            }else {
                log.error("获取失败，状态码:" + response.getStatusLine().getStatusCode());
                return "";
            }
        }catch (Exception ex){
            log.error("请求"+apiPath+"失败");
            throw ex;
        }finally {
            if (response != null) {
                response.close();
            }
            //相当于关闭浏览器
            httpclient.close();
        }
    }
    /**
     * 解析res,生成sql
     * @param res response返回结果 json字符串
     * @param table_name
     * @return  sql
     */
    public static String handlerResToGetSql(String res,String table_name){
        //判断fieldNames是否已经拼接好了
        Boolean fieldNamesState = false;
        String resSql = "insert into "+ table_name;
        StringBuilder fieldNames = new StringBuilder("(");
        StringBuilder fieldValues = new StringBuilder();
        JSONObject parseObject = JSONObject.parseObject(res);
        Map<String, Object> innerMap = parseObject.getInnerMap();
        JSONObject o = (JSONObject)innerMap.get("data");
        Integer totalNum = (Integer)o.get("totalNum");
        JSONArray rows = (JSONArray)o.get("rows");
        for(int i = 0; i < rows.size(); i++) {
            StringBuffer tempValue = new StringBuffer("(");
            JSONObject obj = rows.getJSONObject(i);
            Set<String> keys = obj.keySet();
            Iterator<String> iterator = keys.iterator();
            while (iterator.hasNext()){
                String fieldName = iterator.next();
                if(!fieldNamesState){
                    fieldNames.append("\""+fieldName+"\",");
                }
                Object fieldValue = obj.get(fieldName);
                if(fieldValue == null ||  fieldValue instanceof Number){
                    if(fieldValue instanceof BigDecimal){
                        fieldValue = ((BigDecimal) fieldValue).toPlainString();
                    }
                    tempValue.append(fieldValue+",");
                }else {
                    tempValue.append("'"+fieldValue+"',");
                }
            }
            if(!fieldNamesState){
                fieldNamesState = true;
            }
            fieldValues.append(tempValue.substring(0, tempValue.length()-1) + "),");
        }
        String fieldNamesRes = fieldNames.substring(0, fieldNames.length()-1) + ")";
        String fieldValuesRes = fieldValues.substring(0, fieldValues.length()-1) + ";";

        return resSql + fieldNamesRes + " values" + fieldValuesRes;
    }
    /**
     * 任务类
     */
    class Task implements Callable<Boolean>{
        private String params;
        public Task(String params) {
            this.params = params;
        }

        @Override
        public Boolean call() throws Exception {
            //1.解析出模式和表名称
            String  tableName = "";
            String[] split = params.split(",");
            if (split.length == 1) {
                tableName = split[0];
            } else {
                log.error("参数错误,请检查参数数量");
                throw new RuntimeException("参数错误,请检查参数数量");
            }
            //2.获取表的api相关信息
            String getApiInfoSql = "select * from " + api_info_mode + ".api_info where table_name = ?";
            String[] args = new String[]{tableName};
            Map<String, Object> apiInfoMap = null;
            try {
                apiInfoMap = jdbcTemplate.queryForMap(getApiInfoSql, args);
            } catch (DataAccessException e) {
                log.error(e.getMessage(), e);
                throw e;
            }
            System.out.println("stringObjectMap = " + apiInfoMap);
            //3. 执行前
            Object beforeSql = apiInfoMap.get("before_sql");
            if(!Objects.isNull(beforeSql) && ((String)beforeSql).length() > 0){
                try {
                    jdbcTemplate.update((String) beforeSql);
                    log.info("执行前语句:"+beforeSql+"执行成功");
                } catch (DataAccessException e) {
                    log.info("执行前语句:"+beforeSql+"执行失败\n",e);
                    throw e;
                }
            }
            //4. 执行
            ApiInfoDto apiInfoDto = new ApiInfoDto()
                    .setTableName((String) apiInfoMap.get("table_name"))
                    .setApiPath((String) apiInfoMap.get("api_path"))
                    .setPageSize(2000);
            distributeHttpGet(apiInfoDto);
            //5. 执行后
            Object afterSql = apiInfoMap.get("after_sql");
            if(!Objects.isNull(afterSql) && ((String)afterSql).length() > 0){
                try {
                    jdbcTemplate.update((String)afterSql);
                    log.info("执行后语句:"+afterSql+"执行成功");
                } catch (DataAccessException e) {
                    log.info("执行后语句:"+afterSql+"执行失败\n",e);
                    throw e;
                }
            }
            return true;
        }
    }
    /**
     * 分发任务类
     */
    class DistributeTask implements  Callable<TaskReturn>{
        private CountDownLatch mainThreadLatch;
        private CountDownLatch workerThreadLatch;
        private AtomicBoolean rollbackFlag;
        private ApiInfoDto apiInfoDto;
        public DistributeTask(CountDownLatch mainThreadLatch, CountDownLatch workerThreadLatch,AtomicBoolean rollbackFlag,ApiInfoDto apiInfoDto) {
            this.mainThreadLatch = mainThreadLatch;
            this.workerThreadLatch = workerThreadLatch;
            this.rollbackFlag = rollbackFlag;
            this.apiInfoDto = apiInfoDto;
        }
        @Override
        public TaskReturn call() throws Exception {
            // 开启一个事务
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物隔离级别，开启新事务，这样会比较安全些。
            TransactionStatus status = transactionManager.getTransaction(def); // 获得事务状态
            System.out.println("TransactionStatus"+status.toString());
            //封装请求参数
            HashMap<String, String> requestParams = new HashMap<>();
            requestParams.put("pageNumber",apiInfoDto.getPageNum()+"");
            requestParams.put("pageSize",apiInfoDto.getPageSize()+"");
            //发送请求
            String res = httpGet(apiInfoDto.getApiPath(), requestParams);
            //解析出sql
            String sql = handlerResToGetSql(res,apiInfoDto.getTableName());
            long start = System.currentTimeMillis(),end = 0,update = 0;

            //执行sql
            try {
                update = jdbcTemplate.update(sql);
                end = System.currentTimeMillis();
            } catch (DataAccessException e) {
                 //执行失败,设置回滚标志,让其他子线程也回滚
                 rollbackFlag.set(true);
                 //回滚当前任务
                 transactionManager.rollback(status);
                 return new TaskReturn().setCode(500)
                         .setMes("插入第"+requestParams.get("pageNumber")+"页数据失败")
                         .setException(e);
            }finally {
                //子任务完成 无论成功还是失败,都需要减少标志数量
                workerThreadLatch.countDown();
            }
            //等待所有子任务完成
            mainThreadLatch.await();
            //判断回滚标志,为true代表其它一个或多个子任务失败
            if(rollbackFlag.get()){
                transactionManager.rollback(status);
                return new TaskReturn().setCode(400)
                        .setMes("其它子任务失败,导致失败");
            }
            //提交事务
            transactionManager.commit(status);
            return new TaskReturn().setCode(200)
                    .setMes("插入表"+apiInfoDto.getTableName()+"+第"+requestParams.get("pageNumber")+"页总共"+update+"数据,成功，耗时"+ (end - start) / 1000 + "s");
        }
    }


    /**
     * 任务返回类的包装对象
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Accessors(chain = true)
    class TaskReturn{
        private Integer code;
        private String mes;
        private Exception exception;
    }
}



